package drds.data_migrate;


import drds.common.$;
import drds.common.Author;
import drds.common.Else;
import drds.data_migrate.util.Jdbc;
import drds.data_migrate.vo.JobConfig;
import drds.data_migrate.vo.TaskConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;

import java.math.BigInteger;
import java.sql.*;
import java.util.ArrayList;
import java.util.List;

@Author(name = Authors.LI_YANG)
@Slf4j
public class TableSplits

{


    public static final String primary_key_type_string = "string";
    public static final String primary_key_type_long = "long";
    //
    //
    public static String query_sql_template_without_where = "select %s from %s ";
    public static String query_sql_template = "select %s from %s where (%s)";

    public static List<TaskConfig> split(JobConfig jobConfig, int adviceTaskSize) throws SQLException
    {


        //
        String tableName = jobConfig.getTableName();
        String columnNamesString = jobConfig.getColumnNamesString();
        String splitColumnName = jobConfig.getSplitColumnName();
        String where = jobConfig.getWhere();
        boolean hasWhere = $.isNotNullAndNotEmpty(where);


        Pair<Object, Object> minMaxValuePair = getMinMaxValuePair(jobConfig);
        jobConfig.setQuerySql(buildQuerySql(tableName, columnNamesString, where));
        //
        boolean isStringType = primary_key_type_string.equals(jobConfig.getPrimaryKeyType());
        boolean isLongType = primary_key_type_long.equals(jobConfig.getPrimaryKeyType());


        List<String> rangeList;
        if (isLongType)
        {
            rangeList = RangeSplits.getRangeConditionList(//
                    new BigInteger(minMaxValuePair.getLeft().toString()),//
                    new BigInteger(minMaxValuePair.getRight().toString()),//
                    adviceTaskSize, splitColumnName);//
        } else if (isStringType)
        {
            rangeList = RangeSplits.getRangeConditionList(//
                    String.valueOf(minMaxValuePair.getLeft()),//
                    String.valueOf(minMaxValuePair.getRight()), //
                    adviceTaskSize,//
                    splitColumnName, //
                    "'");//
        } else
        {
            throw new Else();
        }
        //
        String querySql;
        List<String> querySqlList = new ArrayList<String>();
        List<TaskConfig> taskConfigList = new ArrayList<TaskConfig>();
        //
        if ($.isHasData(rangeList))
        {
            for (String range : rangeList)
            {
                TaskConfig taskConfig = new TaskConfig();

                querySql = buildQuerySql(tableName, columnNamesString, where) + (hasWhere ? " and " : " where ") + range;

                querySqlList.add(querySql);
                taskConfig.setQuerySql(querySql);
                taskConfigList.add(taskConfig);
            }
        } else
        {
            TaskConfig taskConfig = new TaskConfig();
            querySql = buildQuerySql(tableName, columnNamesString, where) + (hasWhere ? " and " : " where ") + String.format(" %s IS NOT NULL", splitColumnName);
            querySqlList.add(querySql);
            taskConfig.setQuerySql(querySql);
            taskConfigList.add(taskConfig);
        }

        // deal pk is null
        TaskConfig taskConfig = new TaskConfig();
        querySql = buildQuerySql(tableName, columnNamesString, where) + (hasWhere ? " and " : " where ") + String.format(" %s IS NULL", splitColumnName);

        querySqlList.add(querySql);
        taskConfig.setQuerySql(querySql);
        taskConfigList.add(taskConfig);

        return taskConfigList;
    }

    public static String buildQuerySql(String tableName, String columnName, String where)
    {
        String querySql;
        if ($.isNullOrEmpty(where))
        {
            querySql = String.format(query_sql_template_without_where, columnName, tableName);
        } else
        {
            querySql = String.format(query_sql_template, columnName,
                    tableName, where);
        }
        return querySql;
    }

    //
    @SuppressWarnings("resource")
    private static Pair<Object, Object> getMinMaxValuePair(JobConfig jobConfig) throws SQLException
    {


        String jdbcUrl = jobConfig.getJdbcUrl();
        String username = jobConfig.getUsername();
        String password = jobConfig.getPassword();
        //
        String tableName = jobConfig.getTableName();
        String splitColumnName = jobConfig.getSplitColumnName();
        String where = jobConfig.getWhere();
        //
        Connection connection = Jdbc.getConnection(jdbcUrl, username, password);
        //
        String splitColumnRangeSql = generateSplitColumnRangeSql(tableName, splitColumnName, where);
        Pair<Object, Object> minMaxValuePair = getMinMaxValuePair(jobConfig, connection, splitColumnRangeSql);
        Jdbc.close(null, null, connection);
        return minMaxValuePair;
    }

    public static void getMinMaxValuePair(Connection conn, String splitKeyRangeSql) throws SQLException
    {
        Pair<Object, Object> minMaxValuePair = getMinMaxValuePair(null, conn, splitKeyRangeSql);
        if (null == minMaxValuePair)
        {
            throw new IllegalArgumentException("");
        }
    }

    /**
     * create [min max]sql
     */
    public static String generateSplitColumnRangeSql(String tableName, String splitColumnName, String where)
    {
        String sql = String.format("select min(%s),max(%s) from %s", splitColumnName, splitColumnName, tableName);
        if ($.isNotNullAndNotEmpty(where))
        {
            sql = String.format("%s where %s)", sql, where);
        }
        return sql;
    }

    /**
     * 检测splitPk的配置是否正确。
     * configuration为null, 是precheck的逻辑，不需要回写PK_TYPE到configuration中
     */
    private static Pair<Object, Object> getMinMaxValuePair(JobConfig jobConfig, Connection connection, String splitColumnRangeSql) throws SQLException
    {

        ResultSet resultSet = null;
        Pair<Object, Object> minMaxValuePair = null;
        try
        {
            resultSet = Jdbc.executeQuery(connection, splitColumnRangeSql);
            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
            if (columnIsValid(resultSetMetaData))
            {
                if (isLongType(resultSetMetaData.getColumnType(1)))
                {
                    if (jobConfig != null)
                    {
                        jobConfig.setPrimaryKeyType(primary_key_type_long);
                    }

                    if (resultSet.next())
                    {
                        minMaxValuePair = new ImmutablePair<Object, Object>(resultSet.getString(1), resultSet.getString(2));
                    }
                } else if (isStringType(resultSetMetaData.getColumnType(1)))
                {
                    if (jobConfig != null)
                    {
                        jobConfig.setPrimaryKeyType(primary_key_type_string);
                    }
                    if (resultSet.next())
                    {
                        minMaxValuePair = new ImmutablePair<Object, Object>(resultSet.getString(1), resultSet.getString(2));
                    }
                } else
                {
                    throw new IllegalArgumentException("类型错误");
                }
            } else
            {
                throw new IllegalArgumentException("类型错误");
            }
        } catch (Exception e)
        {
            e.printStackTrace();
            log.error(e.getMessage());
            throw new RuntimeException(e);
        } finally
        {
            Jdbc.close(resultSet, null, null);
        }

        return minMaxValuePair;
    }

    private static boolean columnIsValid(ResultSetMetaData resultSetMetaData) throws SQLException
    {
        int minType = resultSetMetaData.getColumnType(1);
        int maxType = resultSetMetaData.getColumnType(2);
        boolean isNumberType = isLongType(minType);
        boolean isStringType = isStringType(minType);
        if (minType == maxType && (isNumberType || isStringType))
        {
            return true;
        } else
        {
            return false;
        }
    }


    private static boolean isLongType(int type)
    {
        return type == Types.BIGINT ||//
                type == Types.INTEGER ||//
                type == Types.SMALLINT || //
                type == Types.TINYINT;//

    }

    private static boolean isStringType(int type)
    {
        return type == Types.CHAR || type == Types.NCHAR || //
                type == Types.VARCHAR || type == Types.NVARCHAR ||//
                type == Types.LONGVARCHAR;//

    }


}